package com.deng.hadoop.hdfs;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IOUtils;

import java.io.*;
import java.net.URI;
import java.net.URISyntaxException;

/**
 * Created By DengLibin
 * Date: 19-4-3
 * Time: 下午12:08
 * Desc:
 */
public class HdfsClient {

    FileSystem fs;
    Configuration conf;

    // 方法：本地没有hadoop系统，但是可以远程访问。根据给定的URI和用户名，访问hdfs的配置参数
    // 此时的conf不需任何设置，只需读取远程的配置文件即可。

    /**
     * 连接文件系统
     * @throws IOException
     * @throws URISyntaxException
     * @throws InterruptedException
     */

    public  HdfsClient() throws IOException, URISyntaxException, InterruptedException {
        conf = new Configuration();
        // Hadoop的用户名
        String hdfsUserName = "root";
        URI hdfsUri = null;
        // HDFS的访问路径
        hdfsUri = new URI("hdfs://192.168.0.103");
        // 根据远程的NN节点，获取配置信息，创建HDFS对象
        fs = FileSystem.get(hdfsUri, conf, hdfsUserName);

    }

    /**
     * 关闭文件系统
     * @throws IOException
     */

    public  void close() throws IOException {
        fs.close();
    }

    /**
     * 创建文件夹
     * @throws IOException
     */

    public  void testMkdir() throws IOException {
        Path path = new Path("/demo1");
        if(fs.exists(path)){
            fs.delete(path, true);
            System.out.println("删除了目录"+path.getName());
        }
        boolean re = fs.mkdirs(path);
        System.out.println("创建目录结果："+re);
    }

    /**
     * 上传文件
     */

    public  void testUpload() throws IOException {
        Path  outPath = new Path("/demo1/WebStorm-2018.2.7.tar.gz");
        FSDataOutputStream outputStream = fs.create(outPath);

        InputStream inputStream = new FileInputStream(new File("/home/denglibin/software/WebStorm-2018.2.7.tar.gz"));

        IOUtils.copyBytes(inputStream, outputStream, conf, true);
    }

    /**
     * 查询文件块信息
     */

    public  void testBlocks() throws IOException {
        Path  path = new Path("/demo1/WebStorm-2018.2.7.tar.gz");
        FileStatus fileStatus = fs.getFileStatus(path);
        BlockLocation[] fileBlockLocations = fs.getFileBlockLocations(path, 0, fileStatus.getLen());
        //数组长度就是块的数量 （每个块有两个副本，供有三份，块信息包括文件字节的开始字节数和字节大小） 每个块默认大小128M
        for(BlockLocation location : fileBlockLocations){
            System.out.println(location);
        }
    }

    /**
     * 下载文件
     */

    public  void testDownload() throws IOException {
        //打开输入流
        FSDataInputStream inputStream = fs.open(new Path("/demo1/hello.txt"));
        //输入流
        OutputStream outputStream = new FileOutputStream(new File("/home/denglibin/hello_download.txt"));
        //复制
        IOUtils.copyBytes(inputStream, outputStream,1024, true);
    }
}
